Streaming Sources & Sinks

Structured Streaming treats real-time data streams as an unbounded table that is being continuously appended. The core API uses spark.readStream to ingest data from continuous Sources and df.writeStream to write output to streaming Sinks.

graph LR
    subgraph Ingestion["1. Streaming Sources"]
        direction TB
        S1["Kafka Topic Events"]
        S2["File Watcher Folder"]
    end
    subgraph Engine["2. Incremental Processing Engine"]
        E1["Continuous Query DSL"]
        E2["Trigger Interval (Trigger)"]
    end
    subgraph Output["3. Streaming Sinks"]
        direction TB
        O1["Storage Sink (Parquet / Delta)"]
        O2["Console Sink (Debugging)"]
    end
    Ingestion --> Engine --> Output
    style Ingestion fill:#eff6ff,stroke:#2563eb,stroke-width:2px;
    style Engine fill:#fff7ed,stroke:#ea580c,stroke-width:2px;
    style Output fill:#f0fdf4,stroke:#16a34a,stroke-width:2px;

Streaming Sources (Ingestion)

A Source represents the system feeding data into Spark:

Streaming Sinks (Storage)

A Sink represents the target storage where computed streams are written:

PySpark Code Example: File-Watcher Stream to Console

Here is a complete script demonstrating how to monitor a directory for new JSON file dumps and stream the results directly to the console:

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, DoubleType

# 1. Setup Spark
spark = SparkSession.builder \
    .appName("Streaming Sources and Sinks") \
    .master("local[*]") \
    .getOrCreate()

# 2. Define schema explicitly for the incoming file stream
# Streaming file sources require an explicit schema declaration!
file_schema = StructType([
    StructField("device_id", StringType(), False),
    StructField("temperature", DoubleType(), True),
    StructField("status", StringType(), True)
])

# 3. Initialize File-Watcher Read Stream
# Spark will watch the 'input stream directory' folder continuously
streaming_df = spark.readStream \
    .format("json") \
    .schema(file_schema) \
    .option("maxFilesPerTrigger", 1) \
    .load("input_stream_directory")

# 4. Filter incoming records
alert_df = streaming_df.filter(F.col("temperature") > 80.0)

# 5. Write Stream to Console Sink (for testing)
# We specify 'checkpointLocation' to track micro-batch offsets for recovery
query = alert_df.writeStream \
    .format("console") \
    .outputMode("append") \
    .option("checkpointLocation", "temp_checkpoints") \
    .start()

# 6. Keep the stream active until terminated
query.awaitTermination()